package com.zyh.flink.day04;

import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

public class CommentBroadcastStateJob1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> keywordsStream = environment.socketTextStream("hadoop10", 9991);
        DataStreamSource<String> commentsStream = environment.socketTextStream("hadoop10", 9992);

        //1 广播低速流
        /*
        * 广播低速流,需要接受一个MapStateDescriptor,广播状态本质上属于MapState
        * 向广播状态中添加的键值对
        * */
        MapStateDescriptor<String, String> msd = new MapStateDescriptor<>("keywords", Types.STRING, Types.STRING);
        BroadcastStream<String> broadcast = keywordsStream.broadcast(msd);
        //2 高速流连接广播流,形成广播连接流
        BroadcastConnectedStream<String, String> broadcastConnectedStream = commentsStream.connect(broadcast);
        //3 基于广播连接流调用process算子
        SingleOutputStreamOperator<String> processStream = broadcastConnectedStream.process(new MyBroadcastProcessFunction1(msd));
        processStream.print();
        environment.execute("BroadcastStateJob");
    }
}
/*
* IN1 高速流的元素类型
* IN2 低速流的元素类型
* OUT 输出结果类型
* */
class MyBroadcastProcessFunction1 extends BroadcastProcessFunction<String,String,String>{
    private MapStateDescriptor<String,String> msd;
    MyBroadcastProcessFunction1(MapStateDescriptor<String,String> msd){
        this.msd = msd;
    }
    /*
    * comment:表示高速流上的一条评论
    * readOnlyContext:获取广播状态(只读)
    * collector:向下游传递数据
    * */
    @Override
    public void processElement(String comment, ReadOnlyContext readOnlyContext, Collector<String> collector) throws Exception {
        //从状态中获取敏感词完成评论过滤
        //获取状态
        ReadOnlyBroadcastState<String, String> broadcastState = readOnlyContext.getBroadcastState(this.msd);
        String keyword = broadcastState.get("keyword");
        if (keyword!=null){
            comment = comment.replace(keyword,"**");
        }
        //向下游传递结果
        collector.collect(comment);
    }

    /*
    * keyword:表示低速流上的一个敏感词
    * context:可以获取广播状态
    * collector:向下游传递数据
    * */
    @Override
    public void processBroadcastElement(String keyword, Context context, Collector<String> collector) throws Exception {
        //将keyword添加到广播状态中
        //获取广播状态
        BroadcastState<String, String> broadcastState = context.getBroadcastState(msd);
        broadcastState.put("keyword",keyword);
    }
}